//
// Created by root on 24-3-13.
//

#include "pipe.h"
#include "filter_plan.h"
#include "physical_operator_states.hpp"
#include "nest_loop_join_plan.h"
#include "join/radixjoin/physical_radix_join.hpp"
#include "physical_filter.hpp"
#include "planner.hpp"
#include "abstract_plan.h"
#include "seq_scan_plan.h"
#include "projection_plan.h"
#include "physical_project.hpp"
#include "physical_operator.hpp"
#include "physical_table_scan.hpp"
#include "compare_expression.h"
#include "constant_value_expression.h"
#include "filter_condition.hpp"
#include "expression.hpp"
#include "column_value_expression.h"
#include "physical_result_collector.hpp"

namespace DaseX {
/*
 * To build Pipe as a ctx by search plan.
 */
    auto Pipe::BuildPipe(const AbstractPlanNodeRef &plan) -> std::vector<std::shared_ptr<PhysicalOperator>> {
        auto type_ = plan->GetType();

        switch (plan->GetType()) {

            case PlanType::Projection: {
                printf("projection!\n");
                const auto *temp_projection = dynamic_cast<const ProjectionPlanNode *>(plan.get());
                std::vector<int> col_ids_;
                for(auto expression: temp_projection->expressions_) {
                    // TODO(zyy): agg not support, maybe support after long time
                    // the most difficult thing is modify agg in select list in physical, cause not implement
                    auto col_idx =  dynamic_cast<const ColumnValueExpression *>(expression.get())->GetColIdx();
                    col_ids_.emplace_back(col_idx);
                }
                // try
                col_ids_.pop_back();
                // std::shared_ptr<PhysicalProject>
                // build PhysicalProject
                auto phy_proj_op = std::make_shared<PhysicalProject>(PhysicalProject(col_ids_));
//    root = phy_proj_op;
                // TODO(zyy): 注意 此暂定所有的输出结果一定先是一个 proj ，如有其他情况，再议。
                root =  std::make_shared<PhysicalResultCollector>(PhysicalResultCollector(phy_proj_op));

                physical_ops.emplace_back(phy_proj_op);
                // a try
                sink_idxs.emplace_back(physical_ops.size()-1);
                sink_type_.emplace_back(PhysicalOperatorType::PROJECTION);
                printf("physical proj wow!\n");
                // auto PhysicalProject();
                phy_proj_op->children_operator = BuildPipe(plan->children_[0]);
                root->children_operator = {phy_proj_op};
                auto temp_info = TableInfo();
                auto temp_table_1 = std::make_shared<Table>(temp_info);
                auto rc = global_catalog.get_table("t1",temp_table_1);
                if(rc == RC::SUCCESS) {
                    printf("1");
                }
                return {root};
            }

            case PlanType::NestedLoopJoin: {
                printf("NestedLoopJoin\n");

                const auto *temp_join = dynamic_cast<const NestedLoopJoinPlanNode *>(plan.get());
                // all is Equal now
                // can be support by change planNode

                auto left_col_idx = dynamic_cast<const ColumnValueExpression *>(temp_join->predicate_.get()->children_[0].get())->GetColIdx();

                auto right_col_idx = dynamic_cast<const ColumnValueExpression *>(temp_join->predicate_.get()->children_[1].get())->GetColIdx();

                // maybe optimizer will be able to use, then try to select which table to probe,
                // but now, i write hard code for this.
                auto phy_join_op = std::make_shared<PhysicalRadixJoin>(PhysicalRadixJoin(left_col_idx,right_col_idx,
                                                                                         true));

                physical_ops.emplace_back(phy_join_op);

                resource_idxs.emplace_back(physical_ops.size()-1);
                resource_type_.emplace_back(PhysicalOperatorType::NESTED_LOOP_JOIN);

                sink_idxs.emplace_back(physical_ops.size()-1);
                sink_type_.emplace_back(PhysicalOperatorType::NESTED_LOOP_JOIN);


                auto left = BuildPipe(plan->children_[0]);

                phy_join_op->children_operator.insert(phy_join_op->children_operator.end(),left.begin(),left.end());

                auto right = BuildPipe(plan->children_[1]);

                phy_join_op->children_operator.insert(phy_join_op->children_operator.end(),right.begin(),right.end());


                return {phy_join_op};

            }

            case PlanType::SeqScan: {
                printf("seqscan!\n");
                const auto *temp = dynamic_cast<const SeqScanPlanNode *>(plan.get());


                // TODO(zyy): 如果要按照对应 scan 物理算子规则构建，则需要支持完整的查询下推
                //            此处为了验证可行性，暂时使用全表

                printf("table name:%s", temp->table_name_.c_str());

//    auto temp_info = TableInfo();
//    auto temp_table_1 = std::make_shared<Table>(temp_info);
//    auto rc = global_catalog.get_table(temp->table_name_,temp_table_1);
//    auto temp_table_1_pnum = temp_table_1->partition_num;
//    std::vector<int> proj_ids;
//    auto proj_size = temp_table_1->fileds.size();
//    for(int i=0;i<proj_size;i++){
//        proj_ids.push_back(i);
//    }
//    std::vector<std::string> proj_names;
//    for(auto col : temp_table_1->fileds) {
//        proj_names.push_back(col->field_name);
//    }
                std::shared_ptr<Table> temp_table = nullptr;
                auto rc = global_catalog.get_table(temp->table_name_,temp_table);
                auto phy_scan_op = std::make_shared<PhysicalTableScan>(temp_table, 72);
                printf("physical scan wow!\n");
                physical_ops.emplace_back(phy_scan_op);

                resource_idxs.emplace_back(physical_ops.size()-1);
                resource_type_.emplace_back(PhysicalOperatorType::TABLE_SCAN);

                return {phy_scan_op};
            }

            case PlanType::Filter: {
                printf("filter!\n");
                // case the limit of physical operator, now only support constant compare, and constant should in right side.
                // now hard code 1 expression
                const auto *temp_filter = dynamic_cast<const FilterPlanNode *>(plan.get());

                // Just for Test
               //  auto root_expression =

                auto col_idx = dynamic_cast<const ColumnValueExpression *>(temp_filter->predicate_.get()->children_[0].get())->GetColIdx();

                std::shared_ptr<BoundColumnRefExpression> left = std::make_shared<BoundColumnRefExpression>(LogicalType::INTEGER, 0, col_idx, 0);

                auto compare_constant = dynamic_cast<const ConstantValueExpression *>(temp_filter->predicate_.get()->children_[1].get())->val_;

                Value value2(compare_constant);

                std::shared_ptr<BoundConstantExpression> right = std::make_shared<BoundConstantExpression>(value2);

                std::shared_ptr<BoundComparisonExpression> comparisonExpression = std::make_shared<BoundComparisonExpression>(ExpressionTypes::COMPARE_GREATERTHAN);

                comparisonExpression->SetLeft(left);
                comparisonExpression->SetRight(right);

                std::shared_ptr<PhysicalFilter> filter1 = std::make_shared<PhysicalFilter>(comparisonExpression);
                filter1->children_operator = BuildPipe(plan->children_[0]);
                return {filter1};
//                if (compare_type == ComparisonType::GreaterThan ) {
//                    auto phy_filter_op = std::make_shared<PhysicalFilter>(PhysicalFilter(col_idx,ConstantCompareCondition(ConstantCompareType::COMPARE_GREATERTHAN,compare_constant)));
//                    physical_ops.emplace_back(phy_filter_op);
//                    printf("Greater than!\n");
//                    phy_filter_op->children_operator = BuildPipe(plan->children_[0]);
//                    return {phy_filter_op};
//                } else if (compare_type == ComparisonType::Equal ) {
//                    auto phy_filter_op = std::make_shared<PhysicalFilter>(PhysicalFilter(col_idx,ConstantCompareCondition(ConstantCompareType::COMPARE_EQUAL,compare_constant)));
//                } else if (compare_type == ComparisonType::GreaterThanOrEqual) {
//                    auto phy_filter_op = std::make_shared<PhysicalFilter>(PhysicalFilter(col_idx,ConstantCompareCondition(ConstantCompareType::COMPARE_GREATERTHANOREQUALTO,compare_constant)));
//                } else if (compare_type == ComparisonType::LessThan) {
//                    auto phy_filter_op = std::make_shared<PhysicalFilter>(PhysicalFilter(col_idx,ConstantCompareCondition(ConstantCompareType::COMPARE_LESSTHAN,compare_constant)));
//                } else if (compare_type == ComparisonType::LessThanOrEqual) {
//                    auto phy_filter_op = std::make_shared<PhysicalFilter>(PhysicalFilter(col_idx,ConstantCompareCondition(ConstantCompareType::COMPARE_LESSTHANOREQUALTO,compare_constant)));
//                } else if (compare_type == ComparisonType::NotEqual) {
//                    auto phy_filter_op = std::make_shared<PhysicalFilter>(PhysicalFilter(col_idx,ConstantCompareCondition(ConstantCompareType::COMPARE_NOTEQUAL,compare_constant)));
//                } else {
//                    throw DaseX::Exception("error compare type, not support now!\n");
//                }

                }

            }



    }


    auto BuildPipelineGroup() -> int {
        return 1;
    }

}